package jmine.tec.subscription.replicator.jgroups;

import java.net.InetAddress;
import java.net.UnknownHostException;

import jmine.tec.subscription.Publication;
import jmine.tec.subscription.Publisher;
import jmine.tec.subscription.SubscriptionManager;
import jmine.tec.subscription.SubscriptionSettings;
import jmine.tec.subscription.impl.PartialPublication;
import jmine.tec.subscription.replicator.AbstractIdentifiedReplicator;
import jmine.tec.subscription.replicator.AbstractPublicationWrapper;
import jmine.tec.subscription.replicator.PublicationReplicator;

import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.ChannelClosedException;
import org.jgroups.ChannelNotConnectedException;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.Receiver;
import org.jgroups.View;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

/**
 * Implementacao de {@link PublicationReplicator} que usa multicast para publicar sincronizar mensagens. Os nos que fazem parte do mesmo
 * cluster devem usar o mesmo channelName para sincronizar mensagens. Por default, o nome do channel eh o nome do host.
 * 
 * @author takeshi
 */
public class MulticastPublicationReplicator extends AbstractIdentifiedReplicator implements PublicationReplicator, InitializingBean,
        DisposableBean {

    private String channelName = System.nanoTime() + "|" + System.currentTimeMillis();

    private Address localAddress;

    private Channel channel;

    /**
     * C'tor
     */
    public MulticastPublicationReplicator() {
        try {
            this.channelName = InetAddress.getLocalHost().getHostName();
        } catch (UnknownHostException e) {
            this.channelName = "localhost";
        }
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> void synchronizePartialPublication(Publisher<P> publisher, SubscriptionSettings settings,
            PartialPublication<P> publication, SubscriptionManager source) {
        AbstractPublicationWrapper wrapper = this.newPartialPublicationWrapper(publisher, settings, publication);
        this.sendMessage(wrapper);
    }

    /**
     * @param wrapper {@link AbstractPublicationWrapper}
     */
    private void sendMessage(AbstractPublicationWrapper wrapper) {
        try {
            this.channel.send(new Message(null, null, wrapper));
        } catch (ChannelNotConnectedException e) {
            throw new IllegalStateException(e);
        } catch (ChannelClosedException e) {
            throw new IllegalStateException(e);
        }
    }

    /**
     * {@inheritDoc}
     */
    public <P extends Publication> void synchronizePublication(Publisher<P> publisher, SubscriptionSettings settings, P publication,
            SubscriptionManager source) {
        AbstractPublicationWrapper wrapper = this.newPublicationWrapper(publisher, settings, publication);
        this.sendMessage(wrapper);
    }

    /**
     * @param channelName the channelName to set
     */
    public void setChannelName(String channelName) {
        this.channelName = channelName;
    }

    /**
     * {@inheritDoc}
     */
    public synchronized void afterPropertiesSet() throws Exception {
        this.channel = new JChannel();
        this.channel.setReceiver(new Receiver() {

            private volatile byte[] state;

            public byte[] getState() {
                return this.state;
            }

            public void receive(Message msg) {
                if (!msg.getSrc().equals(MulticastPublicationReplicator.this.localAddress)) {
                    AbstractPublicationWrapper wrapper = (AbstractPublicationWrapper) msg.getObject();
                    MulticastPublicationReplicator.this.handle(wrapper);
                }
            }

            public void setState(byte[] state) {
                this.state = state;
            }

            public void block() {
            }

            public void suspect(Address suspectedMbr) {
            }

            public void viewAccepted(View newView) {
            }

        });
        this.channel.connect(this.channelName);
        this.localAddress = this.channel.getLocalAddress();
        LOGGER.info("connected to cluster:  " + this.channel.getClusterName() + " on address: " + this.localAddress);
    }

    /**
     * {@inheritDoc}
     */
    public void destroy() throws Exception {
        this.channel.close();
    }

}
